package com.ruyuan.seckill.service.impl;

import com.ruyuan.seckill.cache.Cache;
import com.ruyuan.seckill.domain.enums.CachePrefix;
import com.ruyuan.seckill.domain.enums.TradeErrorCode;
import com.ruyuan.seckill.domain.vo.TradeVO;
import com.ruyuan.seckill.exception.ServiceException;
import com.ruyuan.seckill.service.GoodsStockLockManager;
import com.ruyuan.seckill.service.TradeIntodbManager;
import com.ruyuan.seckill.utils.DateUtil;
import com.ruyuan.seckill.utils.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 交易入库业务管理组件实现类
 */
@Service
@Slf4j
public class TradeIntodbManagerImpl implements TradeIntodbManager {

    private static LinkedBlockingQueue<Runnable> asyncDeductStockThreadPoolQueue =
            new LinkedBlockingQueue<Runnable>(1000);

    /**
     * 发送消息到MQ的线程池
     */
    private static final ExecutorService executorService = new ThreadPoolExecutor(
        Runtime.getRuntime().availableProcessors(),
        Runtime.getRuntime().availableProcessors(),
        1000 * 60,
        TimeUnit.MILLISECONDS,
        asyncDeductStockThreadPoolQueue,
        new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "AsyncDeductStockExecutor_" + this.threadIndex.incrementAndGet());
            }
        });

    private Integer orderCacheTimeout = 60 * 60;
    /**
     * 商品库存管理组件
     */
    @Autowired
    private GoodsStockLockManager goodsStockLockManager;
    /**
     * string类型的redis缓存的操作模板
     */
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    /**
     * 缓存操作工具
     */
    @Autowired
    private Cache cache;

    /**
     * redis库存扣减
     *
     * @param tradeVO 订单信息
     */
    @Override
    public void seckillStockDeduction(TradeVO tradeVO) {
        try {
            // 基于lua脚本的库存扣减
            boolean stockLockResult = goodsStockLockManager.onSeckillRedisStockDeduction(tradeVO);
            if (!stockLockResult) {
                throw new ServiceException(TradeErrorCode.E456.code(), "秒杀失败，请稍后重试");
            }
            // 扣减库存成功 则对当日订单数量进行累加操作
            stringRedisTemplate.opsForValue().increment("order::number::" + DateUtil.startOfTodDay());
            // 压入缓存
            String cacheKey = CachePrefix.TRADE_SESSION_ID_PREFIX.getPrefix() + tradeVO.getTradeSn();
            log.info("订单创建扣减库存的 cacheKey=[{}]", cacheKey);
            // 将交易数据以json字符串的形式放入redis中 之后会将该cacheKey发送至mq 让秒杀订单服务来消费
            this.cache.put(cacheKey, JsonUtil.objectToJson(tradeVO), orderCacheTimeout);
            log.info("订单创建扣减库存的 cacheKey=[{}] 放入的缓存为 [{}]", cacheKey, JsonUtil.objectToJson(tradeVO));
            String o = (String) this.cache.get(cacheKey);
            log.info("cacheKey=[{}] ==> [{}]", cacheKey, JsonUtil.objectToJson(o));
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    // 将cacheKey发送至MQ 执行异步落库操作 达到缓存与数据库的库存一致
                    log.info("发送交易的 cacheKey 到MQ， 通知秒杀订单服务去扣减库存");
                }
            });
        } catch (Exception e) {
            log.error("秒杀下单失败", e);
            throw new ServiceException(TradeErrorCode.E456.code(), "秒杀失败，请稍后重试");

        }
    }


}
